package com.huaxin.dss2.message;


import lombok.extern.slf4j.Slf4j;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.jms.*;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * bsy 厂级连接MQ  改消息模式为 点对点队列
 * @author guoqf
 * @date 2019/9/11
 */
@Component
@Slf4j
public class MessageQueueSubscriber extends Thread {

    private Lock lock = new ReentrantLock();
    private Condition condition = lock.newCondition();

    private String url;
    private String username;
    private String password;
    private String topicName;
    private String clientId;

    private ActiveMQConnectionFactory connectionFactory;
    @Autowired
    private MessageListener messageListener;

    private TopicConnection connection;

    //消息的消费者
    private MessageConsumer messageConsumer;

    //消息的目的地
    private Destination destination;

    public MessageQueueSubscriber() {
    }

    public MessageQueueSubscriber(String url, String username, String password, String topicName, String clientId) {
        this.url = url;
        this.username = username;
        this.password = password;
        this.topicName = topicName;
        this.clientId = clientId;
        this.init();
    }

    private void init(){
        setName(this.getClass().getSimpleName());
        connectionFactory = new ActiveMQConnectionFactory(username, password,url);
        connectionFactory.setAlwaysSyncSend(true);
        start();
    }

    private void closeConnection(){
        if(null!=connection){
            try {
                connection.close();
            } catch (Exception e) {
                log.error("关闭ActiveMQ连接异常");
            }finally {
                connection=null;
            }
        }
    }

    private void connect(){
        lock.lock();
        try {
            connection = connectionFactory.createTopicConnection();
           // connection.setClientID(clientId);
            connection.setExceptionListener(exception -> {
                log.error("mq connection出现异常" + exception.getMessage());
                closeConnection();
                lock.lock();
                condition.signalAll();
                lock.unlock();
            });
            connection.start();
 /*           TopicSession session = connection.createTopicSession(false, Session.AUTO_ACKNOWLEDGE);
            Topic topic = session.createTopic(topicName);
            TopicSubscriber subscriber = session.createSubscriber(topic);
            subscriber.setMessageListener(messageListener);*/

                ///创建session
             Session  session=connection.createSession(false,Session.AUTO_ACKNOWLEDGE);
            //创建一个HelloWorld的消息队列
            destination=session.createQueue(topicName);
            //创建消息消费者
            messageConsumer = session.createConsumer(destination);
            //监听器 有数据进来就开始刷
            messageConsumer.setMessageListener(messageListener);

            log.debug("mq连接成功");
            condition.await();
        } catch (Exception e) {
            log.error("与中间件连接失败",e);
        }finally {
            lock.unlock();
            try {
                Thread.sleep(5000);
            } catch (InterruptedException e) {
            }
        }
    }

    @Override
    public void run() {
        while (true){
            connect();
        }
    }

    public boolean isConnected(){
        return connection!=null;
    }
}
